package com.ryan.demo.jms.tibco;


import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

/**   
 * JmsCunsumerWithSelector
 *  两个消费者，共同监听同一个Queue，但是它们的Selector不同
 * @author 傅心词
 * Date: Mar 7, 2012
 */
public class JmsCunsumerWithSelector
{
    public static void main(String[] args)
    {
        // 一个生产者，2个消费者
        new Thread(new ProducerTopic()).start();
        new Thread(new CunsumerReceiverA()).start();
        new Thread(new CunsumerReceiverB()).start();
    }
}

/**
 * 消息生产者，可以产生receiver='A'/'B'的消息
 * 
 * @author 郭新立
 * 
 */
class ProducerTopic implements Runnable
{
    public void run()
    {
        try
        {
//            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
//                    ActiveMQConnection.DEFAULT_BROKER_URL);
//            Connection connection = connectionFactory.createConnection();
//            connection.start();
//
//            Session session = connection.createSession(Boolean.TRUE,
//                    Session.AUTO_ACKNOWLEDGE);
			Properties pops = new Properties();
			pops.setProperty("java.naming.factory.initial", "com.tibco.tibjms.naming.TibjmsInitialContextFactory");
			pops.setProperty("java.naming.provider.url", "tcp://192.168.164.179:7222");
			Context ctx = new InitialContext(pops);
			ConnectionFactory cf = (ConnectionFactory) ctx.lookup("GenericConnectionFactory");
			Connection connection = cf.createConnection("admin", "admin");
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue("test.queue");

            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            producer.setTimeToLive(1000);

            String[] receivers = { "A", "B" };
            int len = 10;
            for (int i = 0; i < len; i++)
            {
                String receiver = receivers[i % receivers.length];
                String msg = i + " Hello JMS,receiver=" + receiver;
                if (i >= len - 2)
                {
                    msg = "exit,receiver=" + receiver;
                }

                System.out.println("Send message: " + msg);
                Message message = session.createTextMessage(msg);
                message.setStringProperty("receiver", receiver);
                producer.send(message);
                session.commit();
                Thread.sleep(100);
            }
            session.close();
            connection.close();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }
}

/**
 * 消费者A，只接收receiver='A'的消息
 * 
 */
class CunsumerReceiverA implements Runnable
{
    public void run()
    {
        try
        {
			Properties pops = new Properties();
			pops.setProperty("java.naming.factory.initial", "com.tibco.tibjms.naming.TibjmsInitialContextFactory");
			pops.setProperty("java.naming.provider.url", "tcp://192.168.164.179:7222");
			Context ctx = new InitialContext(pops);
			ConnectionFactory cf = (ConnectionFactory) ctx.lookup("GenericConnectionFactory");
			final Connection connection = cf.createConnection("admin", "admin");
			final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			
            connection.start();

            Destination destination = session.createQueue("test.queue");

            // 只接受receiver='A'的消息
            MessageConsumer consumer = session.createConsumer(destination,
                    "receiver='A'");
            consumer.setMessageListener(new MessageListener()
            {

                public void onMessage(Message message)
                {
                    if (message instanceof TextMessage)
                    {
                        try
                        {
                            String msg = ((TextMessage) message).getText();
                            if (msg.indexOf("exit") != -1)
                            {
                                session.close();
                                connection.close();
                            }

                            System.out.println("Cunuumer ReceiverA:" + msg);
                        }
                        catch (JMSException e)
                        {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }
        catch (JMSException e)
        {
            e.printStackTrace();
        } catch (NamingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
}

/**
 * 消费者B，只接收receiver='B'的消息
 * 
 */
class CunsumerReceiverB implements Runnable
{
    public void run()
    {
        try
        {
			Properties pops = new Properties();
			pops.setProperty("java.naming.factory.initial", "com.tibco.tibjms.naming.TibjmsInitialContextFactory");
			pops.setProperty("java.naming.provider.url", "tcp://192.168.164.179:7222");
			Context ctx = new InitialContext(pops);
			ConnectionFactory cf = (ConnectionFactory) ctx.lookup("GenericConnectionFactory");
			final Connection connection = cf.createConnection("admin", "admin");
			final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			
            connection.start();
            
            Destination destination = session.createQueue("test.queue");

            // 只接受receiver='B'的消息
            MessageConsumer consumer = session.createConsumer(destination,
                    "receiver='B'");
            consumer.setMessageListener(new MessageListener()
            {
                public void onMessage(Message message)
                {
                    if (message instanceof TextMessage)
                    {
                        try
                        {
                            String msg = ((TextMessage) message).getText();
                            if (msg.indexOf("exit") != -1)
                            {
                                session.close();
                                connection.close();
                            }

                            System.out.println("Cunsumer ReceiverB:" + msg);
                        }
                        catch (JMSException e)
                        {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }
        catch (JMSException e)
        {
            e.printStackTrace();
        } catch (NamingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
}